package cororpc

import (
	"errors"
	"fmt"
	"go.uber.org/zap"
	"reflect"
)

type HandlerCallbackVoid = func(arg IMessageArg)
type HandlerCallbackRet = func(arg IMessageArg) (error, any)

// RPCServer 作为服务方rpc
type RPCServer struct {
	registerHandlers map[any]any // 注册过的回调函数 <handler唯一id(string或者结构体的reflect.TypeOf()), 上面两个类型之一 >
	CallChan         chan *CallInfo
}

// NewServer 创建一个新的监听通道
func NewServer() *RPCServer {
	return &RPCServer{
		registerHandlers: make(map[any]any),
		CallChan:         make(chan *CallInfo),
	}
}

// RegisterHandler 注册回调
func (s *RPCServer) RegisterHandler(hId any, cb any) {
	msgId := hId
	switch hId.(type) {
	case string:
	default:
		msgId = reflect.TypeOf(hId)
	}
	switch cb.(type) {
	case HandlerCallbackVoid:
	case HandlerCallbackRet:
	default:
		panic(fmt.Sprintf("函数回调类型不对:%T, 不支持", cb))
	}
	if _, ok := s.registerHandlers[msgId]; ok {
		panic(fmt.Sprintf("重复注册回调:%v", hId))
	}
	s.registerHandlers[msgId] = cb
}

// PushPTC 这个是在别的协程调用的,传到RPCServer真正的协程去处理, 注意这个没有返回
func (s *RPCServer) PushPTC(arg IMessageArg) {
	f, ok := s.registerHandlers[arg.GetMsgId()]
	if !ok || f == nil {
		zap.S().Warnf("找不到:%v(%T) 对应的回调,无法处理", arg.GetMsgId(), arg.GetMsgId())
		return
	}
	defer func() {
		recover()
	}()
	ci := &CallInfo{
		callFunc: f,
		Arg:      arg,
	}
	select {
	case s.CallChan <- ci:
	default:
		ci.ret(&ReturnInfo{err: errors.New("发送失败,对方可能已关闭或者满")})
	}
}

// PushRPC 调用方的协程里调用的，传到RPCServer真正监听的协程去处理,有返回
func (s *RPCServer) PushRPC(retChan chan *ReturnInfo, arg IMessageArg, cb TypeRetCallback) {
	if nil == cb {
		// rpc必须要有cb
		zap.S().Warnf("消息:%s(%T) 必须要有回调", arg.GetMsgId(), arg.GetMsgId())
		return
	}
	f, ok := s.registerHandlers[arg.GetMsgId()]
	if !ok || f == nil {
		zap.S().Warnf("找不到:%s(%T) 对应的回调,无法处理,直接返回", arg.GetMsgId(), arg.GetMsgId())
		cb(errors.New(fmt.Sprintf("找不到路由:%T", arg.GetMsgId())), nil)
		return
	}
	ci := &CallInfo{
		Arg:      arg,
		retChan:  retChan,
		cb:       cb,
		callFunc: f,
	}
	select {
	case s.CallChan <- ci:
	default:
		cb(errors.New("发送RPC失败,对方可能已经关闭或者满了"), nil)
	}
}

// OnCall 执行请求 服务方的协程里面调用
func (s *RPCServer) OnCall(ci *CallInfo) {
	defer func() {
		if r := recover(); r != nil {
			ci.ret(&ReturnInfo{
				err: r.(error),
			})
		}
	}()
	switch t := ci.callFunc.(type) {
	case HandlerCallbackVoid:
		t(ci.Arg)
		ci.ret(&ReturnInfo{})
	case HandlerCallbackRet:
		err, val := t(ci.Arg)
		ci.ret(&ReturnInfo{
			err: err,
			ret: val,
		})
	}
}

func (s *RPCServer) Close() {
	close(s.CallChan)
	for ci := range s.CallChan {
		ci.ret(&ReturnInfo{err: errors.New("rpc 已关闭")})
	}
}
